#pragma once

#include <mutex>
#include <unordered_map>
#include <muduo/net/TcpServer.h>
#include <muduo/net/TcpClient.h>
#include <muduo/net/TcpConnection.h>
#include <muduo/net/Buffer.h>
#include <muduo/net/EventLoop.h>
#include <muduo/net/EventLoopThread.h>
#include <muduo/base/CountDownLatch.h>
#include "Details.hpp"
#include "Fields.hpp"
#include "AbstractLayer.hpp"
#include "Message.hpp"

namespace rpc
{
    class MuduoBuffer : public BaseBuffer
    {
    public:
        using ptr = std::shared_ptr<MuduoBuffer>;
        MuduoBuffer(muduo::net::Buffer *buf) : _buf(buf)
        {
        }
        virtual size_t readableSize() override
        {
            return _buf->readableBytes();
        }
        virtual int32_t peekInt32() override
        {
            // muduo 库是一个网络库，从缓存区取出数据，会进行网络字节序到本地字节序的转换；
            // 后续发送数据时，需要手动将本地字节序转成网络序
            return _buf->peekInt32();
        }
        virtual void retrieveInt32() override
        {
            _buf->retrieveInt32();
        }
        virtual int32_t readInt32() override
        {
            return _buf->readInt32();
        }
        virtual std::string retrieveAsString(size_t len) override
        {
            return _buf->retrieveAsString(len);
        }

    private:
        muduo::net::Buffer *_buf;
    };

    class BufferFactory
    {
    public:
        template <class... Args>
        static BaseBuffer::ptr create(Args &&...args)
        {
            return std::make_shared<MuduoBuffer>(std::forward<Args>(args)...);
        }
    };

    // |--Len--|--msgtype--|--idlen--|--id--|--body--|
    class LVProtocol : public BaseProtocol
    {
    public:
        using ptr = std::shared_ptr<LVProtocol>;
        // 判断缓冲区中的数据是否足够填充一条消息 -- 不足以填充一条信息就不再往下运行，但不用打 ELOG
        virtual bool canProcessed(const BaseBuffer::ptr &buf) override
        {
            // DLOG("begin to judge buffer can processed");
            if (buf->readableSize() < len_fields_length) // buf可读字段长度，不足len的字段长度
            {
                // ELOG("buf->readableSize() < len_fields_length");
                return false;
            }

            int32_t len = buf->peekInt32(); // 保存了 msglen ... body 的字段长度
            if ((len + len_fields_length) > buf->readableSize())
            {
                // ELOG("len + len_fields_length > buf->readableSize()");
                return false;
            }

            return true;
        }

        // |--Len--|--msgtype--|--idlen--|--id--|--body--|
        virtual std::string serialize(const BaseMessage::ptr &msg) override
        {
            std::string body = msg->serialize();
            std::string id = msg->id();
            uint32_t idlen = htonl(id.size()); // idlen 已经被转成了网络字节序
            uint32_t msgtype = htonl((int32_t)msg->type());
            uint32_t h_len = msgtype_fields_length + idlen_fields_length + id.size() + body.size(); // 用于提前开辟空间
            uint32_t n_len = htonl(h_len);                                                          // 用于作为数据发送

            std::string str;
            // str.resize(len + len_fields_length); // error
            str.reserve(len_fields_length + h_len); // 提前开好空间
            // str = std::to_string(len) + std::to_string(msgtype) + std::to_string(idlen) + id + body;
            str.append((char *)&n_len, len_fields_length);
            str.append((char *)&msgtype, msgtype_fields_length);
            str.append((char *)&idlen, idlen_fields_length);
            str.append(id);
            str.append(body);

            return str;
        }

        // virtual std::string serialize(const BaseMessage::ptr &msg) override
        // {
        //     DLOG("LVprotocol begin to serialize");
        //     std::string body = msg->serialize();
        //     DLOG("body : %s", body.c_str());
        //     DLOG("BaseMessage::ptr &msg serialize success");
        //     std::string id = msg->id();
        //     DLOG("get id success, %s", id.c_str());
        //     uint32_t idlen = htonl(id.size()); // idlen 已经被转成了网络字节序
        //     DLOG("get idlen success, %d", idlen);
        //     uint32_t msgtype = htonl((int32_t)msg->type());
        //     DLOG("get msgtype success, %d", msgtype);
        //     uint32_t h_len = msgtype_fields_length + idlen_fields_length + id.size() + body.size(); // 用于提前开辟空间
        //     DLOG("get h_len success, %d", h_len);
        //     uint32_t n_len = htonl(h_len);                                                          // 用于作为数据发送
        //     DLOG("get n_len success");

        //     std::string str;
        //     // str.resize(len + len_fields_length); // error
        //     str.reserve(len_fields_length + h_len); // 提前开好空间
        //     // str = std::to_string(len) + std::to_string(msgtype) + std::to_string(idlen) + id + body;
        //     str.append((char *)&n_len, len_fields_length);
        //     str.append((char *)&msgtype, msgtype_fields_length);
        //     str.append((char *)&idlen, idlen_fields_length);
        //     str.append(id);
        //     str.append(body);

        //     DLOG("to return str");
        //     DLOG("str : %s", str.c_str());
        //     return str;
        // }

        // |--Len--|--msgtype--|--idlen--|--id--|--body--|
        virtual bool onMessage(BaseBuffer::ptr &buf, BaseMessage::ptr &msg) override // 从缓冲区中读出消息
        {
            // 调用 onMessage 的时候默认缓冲区中的数据，可以填充一条完整的消息
            int32_t len = buf->readInt32();
            MSG_TYPE msgtype = (MSG_TYPE)buf->readInt32();
            int32_t idlen = buf->readInt32();
            std::string id = buf->retrieveAsString(idlen);
            int32_t body_length = len - msgtype_fields_length - idlen_fields_length - idlen;
            std::string body = buf->retrieveAsString(body_length);
            // 将 msg 实例化成一个具体的消息类型对象
            msg = MessageFactory::create(msgtype);
            if (msg.get() == nullptr)
            {
                ELOG("construct message failed");
                return false;
            }
            // 将body反序列化，保存到msg中
            bool n = msg->unserialize(body);
            if (n == false)
            {
                ELOG("body unserialize failed");
                return false;
            }
            msg->setId(id);
            msg->setMsgType(msgtype);
            return true;
        }

    private:
        const size_t len_fields_length = sizeof(int32_t);     // len 的字段长度
        const size_t msgtype_fields_length = sizeof(int32_t); // msg 的字段长度
        const size_t idlen_fields_length = sizeof(int32_t);   // idlen 的字段长度
    };

    class ProtocolFactory
    {
    public:
        template <class... Args>
        static BaseProtocol::ptr create(Args &&...args)
        {
            return std::make_shared<LVProtocol>(std::forward<Args>(args)...);
        }
    };

    class MuduoConnection : public BaseConnection
    {
    public:
        using ptr = std::shared_ptr<MuduoConnection>;
        MuduoConnection(const muduo::net::TcpConnectionPtr &conn, const BaseProtocol::ptr &protocol)
            : _conn(conn), _protocol(protocol)
        {
        }

        virtual bool send(const BaseMessage::ptr &msg) override
        {
            if (!connected())
                return false;

            // std::string body = msg->serialize();
            //  DLOG("msg_body : %s, msg_body.size() : %ld", body.c_str(), body.size());
            std::string body = _protocol->serialize(msg);
            const char* body_str = body.c_str();
            // DLOG("MuduoConnection::send body : %s", body_str);
            _conn->send(body);
            return true;
        }

        virtual void shutdown() override
        {
            _conn->shutdown();
        }

        virtual bool connected() override
        {
            return _conn->connected();
        }

    private:
        muduo::net::TcpConnectionPtr _conn; // 使用muduo库中的TcpConnection进行通信
        BaseProtocol::ptr _protocol;        // 通信协议
    };

    class ConnectionFactory
    {
    public:
        template <class... Args>
        static BaseConnection::ptr create(Args &&...args)
        {
            return std::make_shared<MuduoConnection>(std::forward<Args>(args)...);
        }
    };

    class MuduoServer : public BaseServer
    {
    private:
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                DLOG("connection establish");
                BaseConnection::ptr muduo_conn = ConnectionFactory::create(conn, _protocol);
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    _conns.insert(std::make_pair(conn, muduo_conn)); // 保存客户端的连接信息
                    // using ConnectionCallback = std::function<void(const BaseConnection::ptr &)>;
                    if (_cb_connection)
                        _cb_connection(muduo_conn);
                }
            }
            else
            {
                ELOG("disconnect");
                BaseConnection::ptr muduo_conn;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _conns.find(conn);
                    if (it == _conns.end())
                    {
                        return;
                    }
                    muduo_conn = it->second;
                    _conns.erase(it);
                }
                // using CloseCallback = std::function<void(const BaseConnection::ptr &)>;
                if (_cb_close)
                    _cb_close(muduo_conn);
            }
        }

        void onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp)
        {
            DLOG("processing messages");
            BaseBuffer::ptr bbp = BufferFactory::create(buf);
            while (1)
            {
                bool n = _protocol->canProcessed(bbp);
                if (n == false)
                {
                    // 数据不足
                    if (bbp->readableSize() > sizeof(int32_t) && bbp->peekInt32() > maxDataSize)
                    {
                        conn->shutdown();
                        ELOG("buffer data is too large");
                        return;
                    }
                    break;
                }
                BaseMessage::ptr msg;
                n = _protocol->onMessage(bbp, msg); // 从缓冲区中读出一条消息
                if (n == false)
                {
                    ELOG("buffer data error");
                    return;
                }
                // ELOG("process msg success");

                BaseConnection::ptr base_conn;
                {
                    std::unique_lock<std::mutex> lock(_mutex);
                    auto it = _conns.find(conn); // 找到与 conn 对应的客户端连接(MuduoConnection)
                    if (it == _conns.end())
                    {
                        conn->shutdown(); // 关闭连接
                        return;
                    }
                    base_conn = it->second;
                }
                // ELOG("already to _cb_message");
                // using MessageCallback = std::function<void(const BaseConnection::ptr &, const BaseMessage::ptr &msg)>;
                if (_cb_message)
                    _cb_message(base_conn, msg);
            }
        }

    public:
        using ptr = std::shared_ptr<MuduoServer>;
        MuduoServer(uint16_t port)
            : _server(&_baseloop, muduo::net::InetAddress("0.0.0.0", port), "MuduoServer", muduo::net::TcpServer::kReusePort)
        {
            _protocol = ProtocolFactory::create();
        }
        virtual void start() override
        {
            _server.setConnectionCallback(std::bind(&MuduoServer::onConnection, this, std::placeholders::_1));
            _server.setMessageCallback(std::bind(&MuduoServer::onMessage, this, std::placeholders::_1,
                                                 std::placeholders::_2,
                                                 std::placeholders::_3));

            _server.start();
            _baseloop.loop();
        }

    private:
        const size_t maxDataSize = (1 << 16);
        muduo::net::TcpServer _server;
        muduo::net::EventLoop _baseloop;
        BaseProtocol::ptr _protocol;
        std::mutex _mutex;
        std::unordered_map<muduo::net::TcpConnectionPtr, BaseConnection::ptr> _conns; // 保存已连接的客户端信息
    };

    class ServerFactory
    {
    public:
        template <class... Args>
        static BaseServer::ptr create(Args &&...args)
        {
            return std::make_shared<MuduoServer>(std::forward<Args>(args)...);
        }
    };

    class MuduoClient : public BaseClient
    {
    private:
        void onConnection(const muduo::net::TcpConnectionPtr &conn)
        {
            if (conn->connected())
            {
                DLOG("connection establish");
                _downlatch.countDown(); // -- , 唤醒线程
                _conn = ConnectionFactory::create(conn, _protocol);
            }
            else
            {
                ELOG("disconnect");
                _conn.reset();
            }
        }

        void onMessage(const muduo::net::TcpConnectionPtr &conn, muduo::net::Buffer *buf, muduo::Timestamp)
        {
            DLOG("processing messages");
            BaseBuffer::ptr base_buf = BufferFactory::create(buf);
            // DLOG("make basebuffer success");
            while (1)
            {
                bool n = _protocol->canProcessed(base_buf);
                if (n == false) // 缓冲区数据不足以填充一条完整的消息
                {
                    if (base_buf->readableSize() > maxDataSize)
                    {
                        conn->shutdown(); // 关闭连接的写端
                        ELOG("buffer data is too large");
                        return;
                    }
                    break;
                }

                BaseMessage::ptr msg;
                n = _protocol->onMessage(base_buf, msg);
                if (n == false)
                {
                    ELOG("buffer data error");
                    return;
                }

                if (_cb_message)
                    _cb_message(_conn, msg);
            }
        }

    public:
        using ptr = std::shared_ptr<MuduoClient>;
        MuduoClient(std::string serverip, uint16_t serverport)
            : _downlatch(1), _baseloop(_loopthread.startLoop()), _protocol(ProtocolFactory::create()), _client(_baseloop, muduo::net::InetAddress(serverip, serverport), "MuduoClient")
        {
        }

        virtual void connect() override
        {
            // 设置连接事件的回调函数
            _client.setConnectionCallback(std::bind(&MuduoClient::onConnection, this, std::placeholders::_1));

            // 设置消息事件的回调函数
            _client.setMessageCallback(std::bind(&MuduoClient::onMessage, this, std::placeholders::_1,
                                                 std::placeholders::_2,
                                                 std::placeholders::_3));

            _client.connect(); // 连接服务器
            _downlatch.wait(); // 将线程阻塞
        }
        virtual bool send(const BaseMessage::ptr &msg) override
        {
            if (!connected())
            {
                ELOG("disconnect");
                return false;
            }
            return _conn->send(msg); // MuduoConnection::send内部会对msg序列化
        }
        // virtual void disconnect() override
        // {
        //     // _conn->shutdown();
        //     _client.disconnect(); // 断开客户端与服务器的连接
        // }

        virtual void shutdown() override
        {
            _conn->shutdown();
            // _client.disconnect(); // 断开客户端与服务器的连接
        }

        virtual bool connected() override
        {
            // return _conn->connected(); // error
            return _conn && _conn->connected();
        }
        virtual BaseConnection::ptr connection() override
        {
            return _conn;
        }

    private:
        const size_t maxDataSize = (1 << 16);
        std::mutex _mutex;
        BaseProtocol::ptr _protocol;
        BaseConnection::ptr _conn;
        // TcpConnection::send 将数据添加到输出缓冲区中，并将发送任务提交到事件循环中；
        // 事件循环负责监控套接字的可写状态，并在条件满足时发送出去
        muduo::CountDownLatch _downlatch;
        muduo::net::EventLoopThread _loopthread;
        muduo::net::EventLoop *_baseloop;
        muduo::net::TcpClient _client;
    };

    class ClientFactory
    {
    public:
        template <class... Args>
        static BaseClient::ptr create(Args &&...args)
        {
            return std::make_shared<MuduoClient>(std::forward<Args>(args)...);
        }
    };
}